Spark SQL - Advanced Transformations: Theoretical Quiz

This assessment details complex join strategies, window partitioning, and dynamic partition pruning limits.

Scenario 1: Spark Join Strategies (BHJ vs. SMJ vs. SHJ)

The Scenario

A data platform architect is optimizing an ETL pipeline joining a massive sales table (fact_sales, 500 million rows) with a customer lookup table (dim_customers, 200,000 rows). The default configuration utilizes a Sort-Merge Join (SMJ), causing heavy network shuffle write spikes.

The Questions

1. Compare Broadcast Hash Join (BHJ) and Sort-Merge Join (SMJ) in terms of network overhead, partition swapping, and memory limitations.
2. How does one trigger a Broadcast Hash Join, what is the default size limit (spark.sql.autoBroadcastJoinThreshold), and what happens if the broadcasted table is too large for the executor memory?

Detailed Solution & Architectural Analysis

1. BHJ vs. SMJ Execution Comparison

• Sort-Merge Join (SMJ):
  • Network: Extremely high. Both tables are hashed on the join key, partitioned, and shuffled across the network so that matching keys end up on the same executor node.
  • Sorting: Each partition must be sorted by the join key before merging, causing heavy CPU and disk-spill overhead.
  • Memory: Safe for joining two extremely massive tables because sorting/merging can spill to disk if RAM is restricted.
• Broadcast Hash Join (BHJ):
  • Network: Zero shuffle of the large table. The small table is serialized, broadcasted over the network, and loaded in memory on every executor.
  • Sorting: No sorting required. The executor builds a local hash map of the small table and scans the large table's local partition in-place.
  • Memory: Small table must fit entirely within the executor JVM's memory heap, otherwise the executor will crash.

  SMJ: [ fact_sales (Shuffle) ]  (Network)  [ dim_customers (Shuffle) ]  (Sort-Merge on Reducer)

  BHJ: [ fact_sales (Local Scan) ]  (In-Memory Join)  [ dim_customers (Broadcasted to all nodes) ]

2. In-Memory Limits & Overflows

• Size Threshold: Default limit is 10 Megabytes (spark.sql.autoBroadcastJoinThreshold = 10485760 bytes).
• Forcing BHJ: Can be forced using broadcast() hints:

from pyspark.sql.functions import broadcast
joined_df = fact_sales.join(broadcast(dim_customers), "customer_id")

• OOM Risk: If the broadcasted table exceeds the available executor execution memory (e.g. 5GB table broadcasted to a 2GB executor JVM), the JVM will fail to instantiate the hash map, throwing a fatal OutOfMemoryError and crashing the application.

Scenario 2: Window Function Partitioning Overheads

The Scenario

An ecommerce report aggregates customer purchase orders: Window.partitionBy("customer_id").orderBy("order_date") There are 50 million unique customer IDs, and each customer has at most 3 orders. The execution logs reveal heavy partition skew and slow execution times.

The Questions

1. Explain how a Window function forces a physical shuffle in Spark.
2. What are the memory risks when a partition key has severe skew (e.g. a corporate buyer customer has 10 million orders)?

Detailed Solution & Architectural Analysis

1. Window Partitioning Shuffle Mechanics

To evaluate a Window function (e.g. running total or ranking), Spark must co-locate all records sharing the partition key (customer_id) into a single physical partition inside one executor JVM.

• Data Routing: Spark routes all matching customer records across the network to the assigned executor.
• In-Partition Sort: Once shuffled, Spark sorts the records sequentially by order_date within that localized partition, calculating the sliding metrics.

2. Skew Risks

If one key (customer_id = 'corp_buyer_1') has millions of records:

• Spark will route all 10 million records to one executor node, while other executors handle tiny 3-row partitions.
• This creates severe data skew. The executor handling the corp buyer must hold and sort millions of records in memory.
• If the sorted block size exceeds the available memory, it will spill heavily to disk or throw a JVM OutOfMemoryError, stalling the entire application.

Scenario 3: Shuffle Hash Join (SHJ) vs. Sort-Merge Join (SMJ)

The Scenario

An enterprise pipeline is joining two tables where SMJ is too slow, but the smaller table is 150MB (exceeding default broadcast limits). The architect considers enabling Shuffle Hash Join.

The Questions

1. Detail the execution steps of a Shuffle Hash Join.
2. What are the architectural trade-offs of Shuffle Hash Join over Sort-Merge Join?

Detailed Solution & Architectural Analysis

1. Shuffle Hash Join Mechanics

1. Shuffle Phase: Both tables are hashed on the join key and shuffled across the network so that matching keys reside on the same executors.
2. In-Memory Hash Table: Unlike SMJ, Spark does not sort the shuffled partitions. Instead, it reads the shuffled partition of the smaller table and constructs a local in-memory hash table.
3. Join Scan: It scans the larger table's partition, looking up matching keys in the hash table to output joined rows.

2. Trade-offs & Rules

• Advantages: Bypasses the expensive CPU-sorting phase required by SMJ, executing much faster when keys are unsorted on disk.
• Disadvantages: Requires the shuffled partition of the smaller table to fit entirely inside the executor memory pool. If partition skew occurs, it will immediately crash with an OOM, whereas SMJ is much safer due to its disk-spilling sorting engine.

Scenario 4: High-Cardinality Pivot Transformations

The Scenario

A reporting pipeline pivots a log table containing 5,000 distinct product SKUs. The query takes 45 minutes to execute.

The Questions

1. Why are high-cardinality pivot() queries highly expensive in Spark SQL?
2. How does providing explicit pivot values improve execution?

Detailed Solution & Architectural Analysis

1. Pivot Execution Bottleneck

When a developer calls df.groupBy("user_id").pivot("sku").agg(sum("amount")) without specifying the SKU list:

• Spark must launch a pre-flight scan of the entire dataset to discover every unique value in the sku column.
• Once discovered, it shuffles and pivots the columns, generating 5,000 new columns inside the execution engine. This massive width results in heavy JVM object allocation and slows down execution.

2. Explicit Pivot Optimization

Providing an explicit list of pivot values bypasses the pre-flight scan:

# Pass explicit SKU values to pivot directly
product_skus = ["SKU_1", "SKU_2", "SKU_3"]
df.groupBy("user_id").pivot("sku", product_skus).agg(sum("amount"))

Spark compiles the plan instantly, reading only the matching columns and avoiding the expensive table listing scan completely.